package com.event.binlog.processor;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.net.url.UrlBuilder;
import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.StrUtil;
import com.event.binlog.iface.IDistributedCacheService;
import lombok.extern.slf4j.Slf4j;
import com.alibaba.fastjson.JSONObject;
import com.event.binlog.component.MysqlSchemaComponent;
import com.event.binlog.iface.IDistributedLockService;
import com.event.binlog.properties.DbBinlogProperties;
import com.event.binlog.task.KeepAliveTask;
import com.event.common.constant.BinlogRowEventConst;
import com.event.common.constant.DataType;
import com.event.common.model.*;
import com.github.shyiko.mysql.binlog.BinaryLogClient;
import com.github.shyiko.mysql.binlog.event.*;
import com.github.shyiko.mysql.binlog.event.deserialization.EventDeserializer;
import com.google.common.eventbus.EventBus;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.boot.autoconfigure.jdbc.DataSourceProperties;
import org.springframework.stereotype.Component;
import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import javax.annotation.Resource;
import java.io.Serializable;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * @Description BinlogProcessor
 * @Author Bing
 * @Date 2025/1/17 17:49
 */
@Slf4j
@Component
@ConditionalOnBean(DbBinlogProperties.class)
public class BinlogProcessor {

    @Resource
    private DataSourceProperties dataSourceProperties;

    @Resource
    private MysqlSchemaComponent mysqlSchemaComponent;

    @Resource(name = BinlogRowEventConst.EVENT_NAME)
    private EventBus eventBus;

    @Resource
    private IDistributedLockService distributedLockService;

    @Resource
    private DbBinlogProperties dbBinlogProperties;

    @Resource
    private Optional<IDistributedCacheService> distributedCacheService;

    private BinaryLogClient client;
    // 表id跟表结构事件映射
    private Map<Long, TableMapEventData> dbTableMap = new ConcurrentHashMap<>();

    /**
     * 初始化client
     * @return
     */
    private BinaryLogClient initClient() {
        BinaryLogClient result = null;
        if(StrUtil.isNotBlank(dataSourceProperties.getUrl())){
            String replaceUrl = dataSourceProperties.getUrl().replace("jdbc:mysql://", "http://");
            UrlBuilder urlBuilder = UrlBuilder.of(replaceUrl);
            String schema = StrUtil.removePrefix(urlBuilder.getPathStr(),"/");
            result = new BinaryLogClient(urlBuilder.getHost(), urlBuilder.getPort(), schema,dataSourceProperties.getUsername(), dataSourceProperties.getPassword());
        }else if(dbBinlogProperties.hasDbConfig()){
            result = new BinaryLogClient(dbBinlogProperties.getHost(), dbBinlogProperties.getPort(), dbBinlogProperties.getDatabase(),dbBinlogProperties.getUsername(), dbBinlogProperties.getPassword());
        }
        if(ObjectUtil.isNull(result)){
            throw new RuntimeException("未配置BinaryLogClient数据源");
        }
        // 开启持久化
        if(isPersistence()){
            Object object = distributedCacheService.get().getValue(BinlogRowEventConst.BINLOGROWEVENT_LAST_POSITION);
            if(ObjectUtil.isNotNull(object)){
                BinlogPosition binlogPosition = (BinlogPosition)object;
                result.setBinlogPosition(binlogPosition.getBinlogPosition());
                result.setBinlogFilename(binlogPosition.getBinlogFilename());
            }
        }
        return result;
    }

    /**
     * 是否持久化
     * @return
     */
    private boolean isPersistence(){
        boolean isPersistence = ObjectUtil.defaultIfNull(dbBinlogProperties.getPersistence(),false);
        if(!isPersistence || !distributedCacheService.isPresent()){
            return false;
        }
        return true;
    }

    /**
     * 启动
     * @throws Exception
     */
    @PostConstruct
    public void start() throws Exception {
        int lockTimeOut = ObjectUtil.defaultIfNull(dbBinlogProperties.getLockTime(),BinlogRowEventConst.EVENT_LOCK_TIMEOUT);
        boolean lock = distributedLockService.lock(BinlogRowEventConst.EVENT_LCOK, lockTimeOut);
        if(!lock){
            log.warn("获取分布式锁失败,请检查锁是否被占用");
            return;
        }
        DbListenConfig dbListenConfig = mysqlSchemaComponent.getDbListenConfig();
        // 初始化client
        client = initClient();
        Executors.newSingleThreadScheduledExecutor().execute(() -> {
            try {
                run(client,dbListenConfig);
            }catch (Exception e){
                log.warn("BinlogProcessor启动异常",e);
            }
        });
    }

    /**
     * 运行
     * @param client
     * @param dbListenConfig
     * @throws Exception
     */
    private void run(BinaryLogClient client,DbListenConfig dbListenConfig) throws Exception {
        // 初始化eventDeserializer
        EventDeserializer eventDeserializer = new EventDeserializer();
        // 兼容性(对象序列化配置)
        eventDeserializer.setCompatibilityMode(
                EventDeserializer.CompatibilityMode.DATE_AND_TIME_AS_LONG,
                EventDeserializer.CompatibilityMode.CHAR_AND_BINARY_AS_BYTE_ARRAY
        );
        client.setEventDeserializer(eventDeserializer);
        client.registerEventListener(event -> {
            try {
                postProcess(event,dbListenConfig);
            }catch (Exception e){
                log.warn("BinlogProcessor数据处理异常",e);
            }
            // 持久化binlog位置
            if(isPersistence()){
                setPosition(event);
            }
        });
        // 启动心跳任务/续期任务
        startKeepAliveTask();
        client.connect();
    }

    /**
     * 处理事件
     * @param event
     * @param dbListenConfig
     */
    private void postProcess(Event event,DbListenConfig dbListenConfig){
        // 表结构监听
        EventData data = event.getData();
        if(event.getHeader().getEventType().equals(EventType.TABLE_MAP)){
            // 初始化TableMapEventData
            checkTableMapAndInit(dbListenConfig,data);
            return;
        }
        // 行数据监听(增删改)
        if(!EventType.isRowMutation(event.getHeader().getEventType())){
            return;
        }
        // 校验行数据
        if(!checkRowMutation(dbListenConfig,data)){
            return;
        }
        // 处理数据
        process(dbListenConfig,data);
    }

    /**
     * 设置binlog位置
     * @param event
     */
    private void setPosition(Event event){
        EventHeaderV4 headerV4 = event.getHeader();
        EventType eventType = headerV4.getEventType();
        if(eventType != EventType.FORMAT_DESCRIPTION) {
            BinlogPosition binlogPosition = null;
            if (EventType.ROTATE == eventType) {
                binlogPosition = new BinlogPosition();
                binlogPosition.setBinlogFilename(((RotateEventData) event.getData()).getBinlogFilename());
                binlogPosition.setBinlogPosition(((RotateEventData) event.getData()).getBinlogPosition());
            }else{
                Object object = distributedCacheService.get().getValue(BinlogRowEventConst.BINLOGROWEVENT_LAST_POSITION);
                if(ObjectUtil.isNotNull(object)){
                    binlogPosition = (BinlogPosition)object;
                    binlogPosition.setBinlogPosition(headerV4.getPosition());
                }
            }
            if(ObjectUtil.isNotNull(binlogPosition)){
                distributedCacheService.get().setValue(BinlogRowEventConst.BINLOGROWEVENT_LAST_POSITION,binlogPosition);
            }
        }
    }

    /**
     * 处理写数据
     * @param tableMap
     * @param tableSchema
     * @param data
     */
    private void doWriteRowsEventData(TableMapEventData tableMapEventData,JSONObject tableMap, List<TableSchema> tableSchema, WriteRowsEventData data){
        for(Serializable[] item : data.getRows()){
            Map<String,Object> beanMap = convert(tableMap,tableSchema,item);
            // 发送事件
            BinglogAddEvent event = new BinglogAddEvent();
            event.setDatabase(tableMapEventData.getDatabase());
            event.setTableName(tableMapEventData.getTable());
            event.setData(beanMap);
            eventBus.post(event);
        }
    }

    /**
     * 处理更新数据
     * @param tableMap
     * @param tableSchema
     * @param data
     */
    private void doUpdateRowsEventData(TableMapEventData tableMapEventData,JSONObject tableMap, List<TableSchema> tableSchema, UpdateRowsEventData data){
        for(Map.Entry<Serializable[], Serializable[]> item : data.getRows()){
            Map<String,Object> beanMap = convert(tableMap,tableSchema,item.getValue());
            Map<String,Object> beforeMap = convert(tableMap,tableSchema,item.getKey());
            // 发送事件
            BinglogUpdateEvent event = new BinglogUpdateEvent();
            event.setDatabase(tableMapEventData.getDatabase());
            event.setTableName(tableMapEventData.getTable());
            event.setData(beanMap);
            event.setBeforeData(beforeMap);
            eventBus.post(event);
        }
    }

    /**
     * 处理删除数据
     * @param tableMap
     * @param tableSchema
     * @param data
     */
    private void doDeleteRowsEventData(TableMapEventData tableMapEventData,JSONObject tableMap,List<TableSchema> tableSchema,DeleteRowsEventData data){
        for(Serializable[] item : data.getRows()){
            Map<String,Object> beanMap = convert(tableMap,tableSchema,item);
            // 发送事件
            BinglogDelEvent event = new BinglogDelEvent();
            event.setDatabase(tableMapEventData.getDatabase());
            event.setTableName(tableMapEventData.getTable());
            event.setData(beanMap);
            eventBus.post(event);
        }
    }

    /**
     * 处理数据
     * @param dbListenConfig
     * @param data
     */
    private void process(DbListenConfig dbListenConfig,EventData data){
        if(data instanceof WriteRowsEventData){
            // 写入数据
            WriteRowsEventData temp = (WriteRowsEventData)data;
            TableMapEventData tableMapEventData = dbTableMap.get(temp.getTableId());
            List<TableSchema> tableSchemaList = dbListenConfig.getTableSchemaMap().get(tableMapEventData.getTable());
            if(tableSchemaList.size() != temp.getIncludedColumns().length()){
                return;
            }
            doWriteRowsEventData(tableMapEventData,dbListenConfig.getTableMap().get(tableMapEventData.getTable()),tableSchemaList,temp);
        }else if(data instanceof DeleteRowsEventData){
            // 删除数据
            DeleteRowsEventData temp = (DeleteRowsEventData)data;
            TableMapEventData tableMapEventData = dbTableMap.get(temp.getTableId());
            List<TableSchema> tableSchemaList = dbListenConfig.getTableSchemaMap().get(tableMapEventData.getTable());
            if(tableSchemaList.size() != temp.getIncludedColumns().length()){
                return;
            }
            doDeleteRowsEventData(tableMapEventData,dbListenConfig.getTableMap().get(tableMapEventData.getTable()),tableSchemaList,temp);
        }else if(data instanceof UpdateRowsEventData){
            // 更新数据
            UpdateRowsEventData temp = (UpdateRowsEventData)data;
            TableMapEventData tableMapEventData = dbTableMap.get(temp.getTableId());
            List<TableSchema> tableSchemaList = dbListenConfig.getTableSchemaMap().get(tableMapEventData.getTable());
            if(tableSchemaList.size() != temp.getIncludedColumns().length()){
                return;
            }
            doUpdateRowsEventData(tableMapEventData,dbListenConfig.getTableMap().get(tableMapEventData.getTable()),tableSchemaList,temp);
        }
    }

    /**
     * 校验TableMap和初始化Map<Long, TableMapEventData> dbTableMap
     * @param dbListenConfig
     * @param data
     * @return
     */
    private boolean checkTableMapAndInit(DbListenConfig dbListenConfig,EventData data){
        TableMapEventData table = (TableMapEventData)data;
        // 监听的库才处理
        if(!dbListenConfig.getDbName().equals(table.getDatabase())){
            return false;
        }
        // 监听的表才处理
        if(!dbListenConfig.getTableMap().containsKey(table.getTable())){
            return false;
        }
        // 获取表的Schema信息
        List<TableSchema> tableSchemaList = dbListenConfig.getTableSchemaMap().get(table.getTable());
        if(CollUtil.isEmpty(tableSchemaList)){
            return false;
        }
        // 数量不一样时重新拉Schema信息
        if(tableSchemaList.size() != table.getColumnMetadata().length){
            mysqlSchemaComponent.reloadSchema(table.getTable());
        }
        // 保存表结构
        dbTableMap.put(table.getTableId(),table);
        return true;
    }

    /**
     * 校验ROW事件
     * @param dbListenConfig
     * @param data
     * @return
     */
    private boolean checkRowMutation(DbListenConfig dbListenConfig,EventData data){
        if(data instanceof WriteRowsEventData){
            WriteRowsEventData temp = (WriteRowsEventData)data;
            return checkSchema(dbListenConfig,temp.getIncludedColumns().length(),temp.getTableId());
        }else if(data instanceof DeleteRowsEventData){
            DeleteRowsEventData temp = (DeleteRowsEventData)data;
            return checkSchema(dbListenConfig,temp.getIncludedColumns().length(),temp.getTableId());
        }else if(data instanceof UpdateRowsEventData){
            UpdateRowsEventData temp = (UpdateRowsEventData)data;
            return checkSchema(dbListenConfig,temp.getIncludedColumns().length(),temp.getTableId());
        }
        return true;
    }

    /**
     * 校验列数量是否一致
     * @param dbListenConfig
     * @param curColumnSize
     * @param tableId
     * @return
     */
    private boolean checkSchema(DbListenConfig dbListenConfig,int curColumnSize,Long tableId){
        TableMapEventData tableMapEventData = dbTableMap.get(tableId);
        if(ObjectUtil.isNull(tableMapEventData)){
            return false;
        }
        List<TableSchema> tableSchemaList = dbListenConfig.getTableSchemaMap().get(tableMapEventData.getTable());
        if(CollUtil.isEmpty(tableSchemaList)){
            return false;
        }
        // 监听的库才处理
        if(!dbListenConfig.getDbName().equals(tableMapEventData.getDatabase())){
            return false;
        }
        // 监听的表才处理
        if(!dbListenConfig.getTableMap().containsKey(tableMapEventData.getTable())){
            return false;
        }
        // 数量不一样时重新拉Schema信息
        if(tableSchemaList.size() != curColumnSize){
            mysqlSchemaComponent.reloadSchema(tableMapEventData.getTable());
        }
        return true;
    }

    /**
     * 将数据转换成Map
     * @param tableSchemas
     * @param values
     * @return
     */
    private Map<String,Object> convert(JSONObject tableMap,List<TableSchema> tableSchemas, Serializable[] values){
        Map<String,Object> map = new HashMap<>();
        for(int i =0 ; i < tableSchemas.size(); i++){
            Serializable value = values[i];
            TableSchema tableSchema = tableSchemas.get(i);
            String columnName = tableSchema.getColumnName();
            if(CollUtil.isNotEmpty(tableMap)){
                columnName = tableMap.getString(tableSchema.getColumnName());
                columnName = StrUtil.isBlank(columnName) ? tableSchema.getColumnName() : columnName;
            }
            if(ObjectUtil.isNull(value)){
                map.put(columnName, null);
                continue;
            }
            // 判断是否是时间类型
            if(DataType.TimeType.isDateTime(tableSchema.getDataType())){
                value = DateUtil.date(Long.parseLong(value.toString()));
            }
            map.put(columnName,value);
        }
        return map;
    }

    @PreDestroy
    public void stop() throws Exception {
        if (client != null) {
            client.disconnect();
        }
    }

    /**
     * 启动续期任务
     * @throws Exception
     */
    private void startKeepAliveTask() throws Exception {
        if(ObjectUtil.isNull(client)){
            return;
        }
        // 创建一个单线程的ScheduledExecutorService实例
        ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(BinlogRowEventConst.INT_ONE);
        // 续期任务
        int renewalTime = ObjectUtil.defaultIfNull(dbBinlogProperties.getRenewalTime(), BinlogRowEventConst.INT_THIRTY);
        KeepAliveTask task = new KeepAliveTask(client,distributedLockService,renewalTime);
        // 续期的一半时间执行一次
        long period = renewalTime/BinlogRowEventConst.INT_TWO;
        if(period <= BinlogRowEventConst.INT_ZERO){
            period = BinlogRowEventConst.INT_THREE;
        }
        scheduler.scheduleAtFixedRate(task, BinlogRowEventConst.INT_THREE, period, TimeUnit.SECONDS);
    }
}